package com.example.window;

import com.example.bean.WaterSenSorFunction;
import com.example.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * Created with IntelliJ IDEA.
 * ClassName: WindowApiDemo
 * Package: com.example.window
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-19
 * Time: 16:01
 */

//窗口
public class WindowApiDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        final SingleOutputStreamOperator<WaterSensor> data = env
                .socketTextStream("hadoop102", 7777)
                .map(new WaterSenSorFunction());

        //keyBY
        KeyedStream<WaterSensor, String> dataBy = data.keyBy(value -> value.getId());

        //窗口编程流程
        //1.指定窗口分配器 时间/计量 滚动 滑动 会话
        //没有keyBY的窗口 所有数据进入一个子任务 并行度只能为1
        //data.windowAll();

        //就会根据key 分为多条数据流
        //每一个key都会开一个窗口

        //基于时间的
        //指定10秒开一个窗口 滚动窗口
        //dataBy.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        //滑动窗口 窗口长度 10s  滑动步长2s
        //dataBy.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(2)));

        //会话窗口 时间定位5秒
        //dataBy.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));

        //基于计数的
        //滑动窗口 窗口长5个元素
        //dataBy.countWindow(5);

        //滑动窗口 窗口长度5 步长是2
        //dataBy.countWindow(5,2);

        //全局窗口 计数窗口的底层就是这个 自定义的 触发器 一般很少用
        //dataBy.window(GlobalWindows.create())

        //返回的是一个窗口的流
        WindowedStream<WaterSensor, String, TimeWindow> sensorWS =
                dataBy.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        //2.编写窗口函数 窗口内数据的计算逻辑 窗口触发时候计算

        //分为两类
        //1.增量聚合 来一条数据 计算一条数据 并不会立马输出 到窗口触发的时候输出结果
        //sensorWS
                //.reduce()
//        .aggregate();
        //2.全窗口函数 数据来了不计算  到窗口触发的时候计算

        //.process() //这个就是全窗口


        env.execute();
    }
}
